package com.swak.boot.config.mq;

import static com.swak.boot.Application.APP_LOGGER;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.rabbitmq.client.Connection;
import com.swak.Constants;
import com.swak.boot.config.customizer.RabbitOptionsCustomizer;
import com.swak.rabbit.EventBus;
import com.swak.rabbit.RabbitMQProperties;
import com.swak.rabbit.RabbitMQTemplate;
import com.swak.rabbit.RabbitMqInitialize;
import com.swak.rabbit.retry.AbstractRetryConsumer;
import com.swak.rabbit.retry.AutoRetryConsumer;
import com.swak.rabbit.retry.AutoRetryPublisher;
import com.swak.rabbit.retry.RetryPublisher;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.threads.ScheduledContext;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.reactivex.threads.WorkerContext;

/**
 * 消息队列的自动化配置
 * 
 * @see 如果只有生产者 请将：notShareConnection 为 false
 * 
 * @author lifeng
 */
@Configuration
@ConditionalOnClass({ Connection.class, RabbitMQTemplate.class })
@EnableConfigurationProperties(RabbitMQProperties.class)
@ConditionalOnProperty(prefix = Constants.APPLICATION_PREFIX, name = "enableMq", matchIfMissing = true)
public class RabbitMqAutoConfiguration {

	@Autowired
	private RabbitMQProperties properties;

	public RabbitMqAutoConfiguration() {
		APP_LOGGER.debug("Loading Rabbitmq");

	}

	/**
	 * 自动注册消费者
	 * 
	 * @return
	 */
	@Bean
	public RabbitMqPostProcessor rabbitMqPostProcessor() {
		return new RabbitMqPostProcessor();
	}

	/**
	 * 自动开启重试: 需要手动开启
	 * 
	 * @return
	 */
	@Bean
	@ConditionalOnMissingBean(AbstractRetryConsumer.class)
	@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "autoRetry", matchIfMissing = false)
	public AutoRetryConsumer autoRetryConsumer() {
		return new AutoRetryConsumer();
	}

	/**
	 * 自动开启重试
	 * 
	 * @return
	 */
	@Bean
	@ConditionalOnMissingBean(RetryPublisher.class)
	public AutoRetryPublisher autoRetryPublisher() {
		return new AutoRetryPublisher();
	}

	/**
	 * 默认的生产者模板， 默认是不需要自动恢复生产者连接
	 * 
	 * @return
	 */
	@Bean
	public RabbitMQTemplate template(ObjectProvider<List<RabbitOptionsCustomizer>> customizersProvider) {
		
		// 基本配置
		RabbitMQTemplate template = new RabbitMQTemplate(properties);
		WorkerContext workers = Contexts.createWorkerContext("RabbitMQ.Consumers-",
				this.properties.getConsumerThreads(), true, 2, TimeUnit.SECONDS);
		ScheduledContext scheduler = Contexts.createScheduledContext("RabbitMQ.Heartbeat-", 1, true, 60,
				TimeUnit.SECONDS);
		template.setConsumerWorkServiceExecutor(workers).setShutdownExecutor(workers)
				.setTopologyRecoveryExecutor(workers).setHeartbeatExecutor(scheduler);

		// 开启NIO - 感觉 NIO 有问题
		if (this.properties.isUseNio()) {
			WorkerContext nios = Contexts.createWorkerContext("RabbitMQ.EventLoop-", 1, true, 2, TimeUnit.SECONDS);
			template.setNioExecutor(nios);
		}

		// 不支持内部创建线程
		template.setDaemonFactory(new SwakThreadFactory("RabbitMQ-Daemons-", true, new AtomicInteger()));
		List<RabbitOptionsCustomizer> customizers = customizersProvider.getIfAvailable();
		if (customizers != null && !customizers.isEmpty()) {
			for (RabbitOptionsCustomizer customizer : customizers) {
				customizer.customize(template);
			}
		}
		return template;
	}

	/**
	 * 注册Event Bus
	 * 
	 * @param templateForSender
	 * @param retryStrategyProvider
	 * @return
	 */
	@Bean
	public EventBus rabbitEventBus(RabbitMQTemplate template, ObjectProvider<RabbitMqInitialize> initializeProvider,
			RetryPublisher retryPublisher) {
		RabbitMqInitialize initialize = initializeProvider.getIfAvailable();
		if (initialize == null) {
			initialize = (t) -> true;
		}
		WorkerContext executor = Contexts.createWorkerContext("RabbitMQ-Publishers-",
				this.properties.getPublisherThreads(), true, 60, TimeUnit.SECONDS);
		return EventBus.builder().setRetryPublisher(retryPublisher).setTemplate(template).setInitialize(initialize)
				.setExecutor(executor).build();
	}
}